package com.zs.main.config.rocketmq;

import com.zs.common.annotation.ConsumerProvider;
import com.zs.service.rocketmq.consumer.DemoRocketMqConsumer;
import io.netty.channel.DefaultChannelId;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.config.SingletonBeanRegistry;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Map;

/**
 * Copyright 2020 ZS Network Co., Ltd. All rights reserved.
 * 本文件仅限于正数网络技术有限公司内部传阅，禁止外泄以及用于其他的商业目的。
 * Created by yuancong on 2022/1/9.
 */
@Slf4j
@Configuration
@ConditionalOnProperty("rocketmq.consumer.namesrvAddr")
public class ConsumerConfig implements ApplicationListener<ContextRefreshedEvent> {
    @Value("${rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.consumeThreadMin:10}")
    private Integer consumeThreadMin;
    @Value("${rocketmq.consumer.consumeThreadMax:32}")
    private Integer consumeThreadMax;
    @Value("${rocketmq.consumer.consumeMessageBatchMaxSize:1}")
    private Integer consumeMessageBatchMaxSize;

    private static Map<String, MQPushConsumer> consumerMap = new HashMap<>();

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        ConfigurableApplicationContext applicationContext = (ConfigurableApplicationContext) event.getSource();
        Map<String, Object> beanMap = applicationContext.getBeansWithAnnotation(ConsumerProvider.class);
        for (Map.Entry<String, Object> entry : beanMap.entrySet()) {
            String beanName = entry.getKey();
            Object bean = entry.getValue();
            ConsumerProvider annotation = bean.getClass().getAnnotation(ConsumerProvider.class);
            String groupName = annotation.groupName();
            String topic = annotation.topic();
            String tag = annotation.tag();
            if (StringUtils.isEmpty(groupName) || StringUtils.isEmpty(topic)
                    || StringUtils.isEmpty(tag)) {
                log.error("ConsumerConfig: groupName or topic or tag is null, {} is ignore",beanName);
                continue;
            }

            if (!(bean instanceof MessageListenerConcurrently)) {
                log.error("ConsumerConfig:  {} is not valid RocketMQ message listener,ignore...",beanName);
                continue;
            }

            DefaultMQPushConsumer consumer = createConsumer(groupName, topic, tag, (MessageListenerConcurrently) bean);
            String consumerName = StringUtils.isEmpty(annotation.name()) ? String.format("mqConsumer%s-%s-%s",groupName,topic,tag) : annotation.name();
            consumerMap.put(consumerName,consumer);
        }
    }

    public static Map<String, MQPushConsumer> getConsumerMap() {
        return consumerMap;
    }

    private DefaultMQPushConsumer createConsumer(String groupName,String topic,String tag,MessageListenerConcurrently listener) {
        log.info("begin to start consumer,groupName={}, topics={},tag={}",groupName,topic,tag);
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        consumer.registerMessageListener(listener);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 设置消费模型，集群还是广播，默认为集群
        // consumer.setMessageModel(MessageModel.CLUSTERING);

        try {
            consumer.subscribe(topic, tag);
            DefaultChannelId.newInstance();
            consumer.start();
            log.info("OK to start consumer ,groupName={}, topics={},tag={}",groupName,topic,tag);
        } catch (MQClientException e) {
            log.error("Failed to start consumer ,groupName={}, topics={},tag={},error message is {}",groupName,topic,tag,e.getErrorMessage());
        }
        return consumer;
    }
}
